package com.atguigu.flink.exec1;

import com.atguigu.flink.function.WaterSensorMapFunction;
import com.atguigu.flink.pojo.OrderEvent;
import com.atguigu.flink.pojo.TxEvent;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.KeyedCoProcessFunction;
import org.apache.flink.util.Collector;

import java.util.HashMap;
import java.util.HashSet;
import java.util.Map;
import java.util.Set;

/**
 * Created by Smexy on 2022/12/14
 *
 *      1.读两种不同的数据，获取两种流
 *      2.两个流 必须 connect才能对账
 *      3.connect之后，保证同一个 支付id(txId)的两种数据可以发到下游的同一个Task
 *              keyBy(txId)
 *      4.connect之后再进行处理，必须调用 coXxx 算子
 *              coXxx 算子在处理时，依旧提供两个方法，处理各自的流。
 *                  每个方法中的变量无法直接访问，必须借助 成员变量(属性)。
 *                      提供两个属性:
 *                              用于缓存OrderEvent
 *                              用于缓存 TxEvent
 *
 *                    如何判断是否对账成功：
 *                              两种流的数据在处理时，读对应的缓存，读到就说明对账成功，输出。
 *                                  读不到，就把自己写入自己的缓存，等待对方类型在处理时，再判断是否可以对账成功。
 */
public class Demo5_OrderPay
{
    public static void main(String[] args) {
        
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        
        //1.获取数据
        SingleOutputStreamOperator<OrderEvent> orderDS = env.readTextFile("data/OrderLog.csv")
                                                        .map(new MapFunction<String, OrderEvent>()
                                                        {
                                                            @Override
                                                            public OrderEvent map(String value) throws Exception {
                                                                String[] words = value.split(",");
                                                                return new OrderEvent(
                                                                    Long.valueOf(words[0]),
                                                                    words[1],
                                                                    words[2],
                                                                    Long.valueOf(words[3])
                                                                );
                                                            }
                                                        })
                                                        .filter(o -> "pay".equals(o.getEventType()));


        SingleOutputStreamOperator<TxEvent> txDS = env.readTextFile("data/ReceiptLog.csv")
                                                     .map(new MapFunction<String, TxEvent>()
                                                     {
                                                         @Override
                                                         public TxEvent map(String value) throws Exception {
                                                             String[] words = value.split(",");
                                                             return new TxEvent(
                                                                 words[0],
                                                                 words[1],
                                                                 Long.valueOf(words[2])
                                                             );
                                                         }
                                                     });

        orderDS.connect(txDS)
               .keyBy(OrderEvent::getTxId,TxEvent::getTxId)
               .process(new KeyedCoProcessFunction<String, OrderEvent, TxEvent, String>()
               {
                   //缓存TxEvent，只需要存txId
                   Set<String> tes = new HashSet<>();

                   //缓存OrderEvent，需要缓存txId,还要缓存orderId
                   Map<String,Long> oes = new HashMap<>();

                   // 处理connect之前的流  orderDS
                   @Override
                   public void processElement1(OrderEvent orderEvent, Context ctx, Collector<String> out) throws Exception {

                       //处理当前的OrderEvent，需要访问 TxEvent的缓存，判断是否对账成功
                       if (tes.contains(orderEvent.getTxId())){
                           out.collect(orderEvent.getOrderId().toString());
                       }else{
                           //要对账的TxEvent还没到达task(要写入缓存，等待)，或压根不存在
                           oes.put(orderEvent.getTxId(),orderEvent.getOrderId());
                       }


                   }

                   // 处理connect之后的流  txDS
                   @Override
                   public void processElement2(TxEvent txEvent, Context ctx, Collector<String> out) throws Exception {
                       Long orderId = oes.get(txEvent.getTxId());

                       if (orderId != null){
                           //说明对账成功了
                           out.collect(orderId.toString());
                       }else{
                           //要对账的OrderEvent还没到达task(要写入缓存，等待)，或压根不存在
                           tes.add(txEvent.getTxId());
                       }

                   }
               })
               .print().setParallelism(1);

        try {
                            env.execute();
                        } catch (Exception e) {
                            e.printStackTrace();
                        }
        
    }
}
